package org.apache.flink.runtime.resourceconsumer.runner;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;

import org.apache.elasticresource.api.BusinessResourceConsumer;
import org.apache.flink.api.common.ConsumerID;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ConsumerManagerOptions;
import org.apache.flink.runtime.heartbeat.HeartbeatServices;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
import org.apache.flink.runtime.leaderelection.LeaderElectionService;
import org.apache.flink.runtime.resourceconsumer.DefaultSlotPoolServiceFactory;
import org.apache.flink.runtime.resourceconsumer.ResourceConsumerConfiguration;
import org.apache.flink.runtime.resourceconsumer.factories.DefaultResourceConsumerServiceFactory;
import org.apache.flink.runtime.resourceconsumer.factories.DefaultResourceConsumerServiceProcessFactory;
import org.apache.flink.runtime.resourceconsumer.factories.ConsumerManagerConsumerMetricGroupFactory;
import org.apache.flink.runtime.resourceconsumer.factories.ResourceConsumerServiceFactory;
import org.apache.flink.runtime.rpc.FatalErrorHandler;
import org.apache.flink.runtime.rpc.RpcService;
import org.apache.flink.runtime.slotpool.SlotPoolServiceFactory;
import org.apache.flink.runtime.util.Hardware;
import org.apache.flink.util.concurrent.ExecutorThreadFactory;

/**
 * @类名: ResourceConsumerServiceLeadershipRunnerFactory
 * @说明: ResourceConsumerServiceRunner领导权管理工厂，构建领导权管理runner
 *
 * @author   leehom
 * @Date	 2023年10月31日 下午2:10:33
 * 修改记录：
 *
 * @see 	 
 */
public enum ResourceConsumerServiceLeadershipRunnerFactory implements ResourceConsumerRunnerFactory {
    INSTANCE;

    @Override
    public ResourceConsumerRunner createResourceConsumerRunner(
    		ConsumerID consumerId,
    		String consumerName,
    		Configuration configuration,
            RpcService rpcService,
            HighAvailabilityServices highAvailabilityServices,
            HeartbeatServices heartbeatServices,
            ConsumerManagerConsumerMetricGroupFactory consumerManagerConsumerMetricGroupFactory,
            FatalErrorHandler fatalErrorHandler,
            long initializationTimestamp,
            BusinessResourceConsumer businessResourceConsumer)
            throws Exception {
    	// 释出资源消费者配置
        final ResourceConsumerConfiguration resourceConsumerConfiguration =
        			ResourceConsumerConfiguration.fromConfiguration(configuration);
        // 主节点选举服务
        final LeaderElectionService resourceConsumerLeaderElectionService =
                highAvailabilityServices.getResourceConsumerLeaderElectionService();
        // slot pool服务工厂
        final SlotPoolServiceFactory slotPoolServiceFactory =
                DefaultSlotPoolServiceFactory.fromConfiguration(configuration);
        // 构建执行器，依据配置
        final int numberCPUCores = Hardware.getNumberCPUCores();
        final int jobManagerIoPoolSize =
        		configuration.getInteger(ConsumerManagerOptions.CONSUMER_MANAGER_IO_POOL_SIZE, numberCPUCores);
        final ExecutorService ioExecutor =
                Executors.newFixedThreadPool(
                        jobManagerIoPoolSize, new ExecutorThreadFactory("resourceconsuemr-io"));
        final int jobManagerFuturePoolSize =
        		configuration.getInteger(ConsumerManagerOptions.JOB_MANAGER_FUTURE_POOL_SIZE, numberCPUCores);
        final ScheduledExecutorService futureExecutor =
                Executors.newScheduledThreadPool(
                        jobManagerFuturePoolSize, new ExecutorThreadFactory("resourceconsuemr-future"));
        // 资源消费者服务工厂        
        final ResourceConsumerServiceFactory resourceConsumerServiceFactory = 
                new DefaultResourceConsumerServiceFactory(
                		consumerId,
                		consumerName,
                		ioExecutor,
                        rpcService,
                        resourceConsumerConfiguration,
                        highAvailabilityServices,
                        slotPoolServiceFactory,
                        heartbeatServices,
                        consumerManagerConsumerMetricGroupFactory,
                        fatalErrorHandler,
                        initializationTimestamp,
                        businessResourceConsumer);
        
        // 资源消费者服务process工厂
        final DefaultResourceConsumerServiceProcessFactory resourceConsumerServiceProcessFactory =
                new DefaultResourceConsumerServiceProcessFactory(
                		consumerId,
                        resourceConsumerServiceFactory);
        
        // 构建资源消费者runner
        return new ResourceConsumerServiceLeadershipRunner(
                resourceConsumerServiceProcessFactory,
                resourceConsumerLeaderElectionService,
                fatalErrorHandler);
    }
}
